package com.sam.kafka.interceptor;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Headers;

import java.nio.charset.StandardCharsets;
import java.util.Map;

/**
 * @author: wm.xue
 * @date: 2021/11/8 13:52
 * @description:
 */
@Slf4j
public class MyInterceptorOne implements ProducerInterceptor {
    @Override
    public ProducerRecord onSend(ProducerRecord record) {
        System.err.println("拦截器1----------go");

        // 此处根据业务需要对相关数据进行修改

        // 添加消息头
        Headers headers = record.headers();
        headers.add("interceptor", "MyInterceptorOne".getBytes(StandardCharsets.UTF_8));


        ProducerRecord newRecord = new ProducerRecord(
                record.topic()
                ,record.partition()
                ,record.timestamp()
                ,record.key()
                ,record.value()
                ,headers);
        return newRecord;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        System.err.println("拦截器1----------back");
        if (exception != null) {
            log.error(exception.getMessage());
        }
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {
    }
}
